package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.data_etl.udfs.GetNull;
import cn.doitedu.rtdw.data_etl.udfs.TimeTruncUdf;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/30
 * @Desc: 学大数据，上多易教育
 **/
public class Job6_广告曝光点击olap分析主题轻度聚合模型 {

    public static void main(String[] args) {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        /**
         * 创建逻辑表（连接器表），映射kafka中物理topic（维度打宽明细行为日志）
         */
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                          "
                        + "     user_id           BIGINT,                     "
                        + "     username          string,                     "
                        + "     session_id        string,                     "
                        + "     event_id          string,                     "
                        + "     event_time        bigint,                     "
                        + "     lat               double,                     "
                        + "     lng               double,                     "
                        + "     release_channel   string,                     "
                        + "     device_type       string,                     "
                        + "     properties        map<string,string>,         "
                        + "     register_phone    STRING,                     "
                        + "     user_status       INT,                        "
                        + "     register_time     TIMESTAMP(3),               "
                        + "     register_gender   INT,                        "
                        + "     register_birthday DATE,                       "
                        + "     register_city        STRING,                  "
                        + "     register_job         STRING,                  "
                        + "     register_source_type INT,                     "
                        + "     gps_province STRING,                          "
                        + "     gps_city     STRING,                          "
                        + "     gps_region   STRING,                          "
                        + "     url_prefix    STRING,                         "
                        + "     page_type    STRING,                          "
                        + "     page_service STRING,                          "
                        + "     proc_time AS proctime(),                      " // processing time 时间语义
                        + "     rt AS  to_timestamp_ltz(event_time,3),        " // 表达式字段，用于将event_time转成timestamp(3)类型
                        + "     WATERMARK FOR rt AS  rt - INTERVAL '0' SECOND " // 基于rt字段定义watermark，从此，rt字段就具备了 flink中事件时间语义
                        + " ) WITH (                                          "
                        + "  'connector' = 'kafka',                           "
                        + "  'topic' = 'dwd_events',                          "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092', "
                        + "  'properties.group.id' = 'testGroup',             "
                        + "  'scan.startup.mode' = 'latest-offset',           "
                        + "  'value.format'='json',                           "
                        + "  'value.json.fail-on-missing-field'='false',      "
                        + "  'value.fields-include' = 'EXCEPT_KEY')           "
        );


        // 抽取广告主题中心事实表数据（ 广告曝光、广告点击、广告转化）
        tenv.executeSql(
                " CREATE TEMPORARY VIEW ad_events AS               " +
                        " SELECT                                            " +
                        "   user_id,                                        " +
                        " 	device_type,                                    " +
                        " 	event_id,                                       " +
                        " 	event_time,                                     " +
                        " 	gps_province,                                   " +
                        " 	properties['ad_region_id'] as ad_region_id,     " +
                        " 	properties['ad_tracking_id'] as ad_tracking_id, " +
                        " 	properties['creative_id'] as creative_id,       " +
                        " 	rt                                              " +
                        " from dwd_kafka                                    " +
                        " where event_id in ('ad_show','ad_click')          "

        );


        //
        //tenv.executeSql("select * from ad_events").print();


        // 按照 用户+广告位+广告创意id  为粒度，聚合 曝光次数、点击次数等
        tenv.createTemporaryFunction("time_trunc", TimeTruncUdf.class);
        tenv.createTemporaryFunction("get_null", GetNull.class);
        tenv.executeSql(
                " create temporary view ad_agg_uid  AS                                                               " +
                        " SELECT                                                                                             " +
                        "   to_date(date_format(window_end,'yyyy-MM-dd'))  as dt,                                          " +
                        " 	time_trunc(window_end,60) as time_60m,                                                           " +
                        " 	time_trunc(window_end,30) as time_30m,                                                           " +
                        " 	time_trunc(window_end,10) as time_10m,                                                           " +
                        " 	time_trunc(window_end,5) as time_5m,                                                             " +
                        " 	time_trunc(window_end,1) as time_1m,                                                             " +
                        "   user_id,                                                                                        " +
                        " 	gps_province,                                                                                    " +
                        " 	device_type,                                                                                     " +
                        " 	ad_region_id,                                                                                    " +
                        " 	creative_id,                                                                                     " +
                        " 	proctime() as pt,                                                                                " +
                        " 	count(1) FILTER(WHERE event_id = 'ad_show') as ad_show_cnt,                                      " +
                        " 	user_id    as ad_show_users,                                                                     " +
                        " 	count(1) FILTER(WHERE event_id = 'ad_click') as ad_click_cnt  ,                                  " +
                        " 	if((count(1) FILTER(WHERE event_id = 'ad_click')>0),user_id,get_null())    as ad_click_users     " +
                        "                                                                                                    " +
                        " FROM TABLE(                                                                                        " +
                        "     TUMBLE(TABLE ad_events,DESCRIPTOR(rt),INTERVAL '1' MINUTE)                                     " +
                        " )                                                                                                  " +
                        " GROUP BY                                                                                           " +
                        "     window_start,                                                                                  " +
                        " 	window_end,                                                                                      " +
                        "     user_id,                                                                                       " +
                        " 	gps_province,                                                                                    " +
                        " 	device_type,                                                                                     " +
                        " 	ad_region_id,                                                                                    " +
                        " 	creative_id                                                                                      "
        );

        // 建表，映射doris中物理目标表
        tenv.executeSql(
                " create table ad_olap_doris  (                                                          "+
                        "    dt             date,                                                                "+
                        "    time_60m       string,                                                              "+
                        "    time_30m       string,                                                              "+
                        "    time_10m       string,                                                              "+
                        "    time_5m        string,                                                              "+
                        "    time_1m        string,                                                              "+
                        "    gps_province   string,                                                              "+
                        "    device_type    string,                                                              "+
                        "    ad_region_id   string,                                                              "+
                        "    creative_id    string,                                                              "+
                        "    ad_id          string,                                                              "+
                        "    advertiser_id  int,                                                                 "+
                        "    ad_campaign    string,                                                              "+
                        "    ad_show_cnt    BIGINT,                                                              "+
                        "    ad_show_users  BIGINT,                                                              "+
                        "    ad_click_cnt   BIGINT,                                                              "+
                        "    ad_click_users BIGINT                                                               "+
                        " ) WITH (                                                                               "+
                        "    'connector' = 'doris',                                                              "+
                        "    'fenodes' = 'doitedu:8030',                                                         "+
                        "    'table.identifier' = 'dws.ad_ana_agg',                                              "+
                        "    'username' = 'root',                                                                "+
                        "    'password' = 'root',                                                                "+
                        "    'sink.label-prefix' = 'doris_label-xxx',                                            "+
                        "    'sink.properties.columns' = 'dt,time_60m,time_30m,time_10m,time_5m,                 "+
                        "    time_1m,gps_province,device_type,ad_region_id,creative_id,ad_id,advertiser_id,      "+
                        "    ad_campaign,ad_show_cnt,ad_show_users,ad_click_cnt,ad_click_users,                   "+
                        "    ad_show_users=to_bitmap(ad_show_users),ad_click_users=to_bitmap(ad_click_users) '   "+
                        " )                                                                                      "

        );


        // 创建一个表，映射hbase中的广告信息维度表
        tenv.executeSql(
                " create table dim_ad_info_hbase(                    "+
                        "   creative_id string,                                    "+
                        " 	f row<ad_id string,advertiser_id STRING,ad_campaign string>,  "+
                        "   primary key(creative_id) not enforced      "+
                        " ) with (                                 "+
                        "     'connector' = 'hbase-2.2',           "+
                        "     'table-name' = 'dim_ad_info',       "+
                        "     'zookeeper.quorum' = 'doitedu:2181'  "+
                        " )                                         "
        );


        // 将上面聚合好的数据，关联 广告信息 维表 ,并插入目标表
        tenv.executeSql(
                " insert into ad_olap_doris                              "+
                        " select                                                 "+
                        "     a.dt                                               "+
                        "     ,a.time_60m                                         "+
                        "     ,a.time_30m                                         "+
                        "     ,a.time_10m                                         "+
                        "     ,a.time_5m                                          "+
                        "     ,a.time_1m                                          "+
                        "     ,a.gps_province                                     "+
                        "     ,a.device_type                                      "+
                        "     ,a.ad_region_id                                     "+
                        "     ,a.creative_id                                      "+
                        "     ,b.ad_id                                            "+
                        "     ,cast(b.advertiser_id  as int) as   advertiser_id   "+
                        "     ,b.ad_campaign                                       "+
                        "     ,a.ad_show_cnt                                      "+
                        "     ,a.ad_show_users                                    "+
                        "     ,a.ad_click_cnt                                     "+
                        "     ,a.ad_click_users                                   "+
                        " from                                                   "+
                        " ad_agg_uid  a                                          "+
                        " left join  dim_ad_info_hbase  for system_time as of a.pt as b    "+
                        " on a.creative_id = b.creative_id                                "
        );



    }

}
